package com.heima.schedule.service.Impl;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.toolkit.StringUtils;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.heima.common.constants.ScheduleConstants;
import com.heima.common.redis.CacheService;
import com.heima.model.schedule.dtos.Task;
import com.heima.model.schedule.pojo.Taskinfo;
import com.heima.model.schedule.pojo.TaskinfoLogs;
import com.heima.schedule.mapper.TaskinfoLogsMapper;
import com.heima.schedule.mapper.TaskinfoMapper;
import com.heima.schedule.service.TaskService;
import com.mysql.cj.log.Log;
import com.mysql.cj.log.LogFactory;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import sun.util.calendar.BaseCalendar;

import javax.annotation.PostConstruct;
import java.util.Calendar;
import java.util.Date;
import java.util.List;
import java.util.Set;

@Service
@Slf4j
public class TaskServiceImpl implements TaskService {
    @Override
    public long addTask(Task task) {

        Boolean success =addTaskToDb(task);

        if (success){
            addTaskToCache(task);
        }



        return task.getTaskId();
    }

    @Override
    public boolean cancelTask(long taskId) {

        boolean flag = false;


        //删除任务  更新任务日志
        Task task = updateDb(taskId,ScheduleConstants.CANCELLED);

        if (task!=null){
            removeTaskFromCache(task);
            flag=true;
        }

        return flag;
    }

    @Override
    public Task poll(int type, int priority) {

        Task task = null;
        try {
            String key = type+"_"+priority;
            String s = cacheService.lRightPop(ScheduleConstants.TOPIC + key);

            if (StringUtils.isNotBlank(s)){
                task = JSON.parseObject(s, Task.class);
                updateDb(task.getTaskId(),ScheduleConstants.EXECUTED);
            }
        }catch (Exception e){
            e.printStackTrace();
            log.error("poll error {}",task.getTaskId());

        }
        return task;
    }

    private void removeTaskFromCache(Task task) {

        String key = task.getTaskType()+"_"+task.getPriority();

        if (task.getExecuteTime()<=System.currentTimeMillis()){
            cacheService.lRemove(ScheduleConstants.TOPIC+key,0,JSON.toJSONString(task));
        }else{
            cacheService.zRemove(ScheduleConstants.FUTURE+key,JSON.toJSONString(task));
        }

    }

    private Task updateDb(long taskId, int status) {

        Task task = null;
        try {

            taskinfoMapper.deleteById(taskId);

            TaskinfoLogs taskinfoLogs = taskinfoLogsMapper.selectById(taskId);
            taskinfoLogs.setStatus(status);
            taskinfoLogsMapper.updateById(taskinfoLogs);

            task = new Task();
            BeanUtils.copyProperties(taskinfoLogs,task);
            task.setExecuteTime(taskinfoLogs.getExecuteTime().getTime());

        }catch (Exception e){

            log.error("task cancelled error");
        }



        return task;

    }

    @Autowired
    private CacheService cacheService;
    private void addTaskToCache(Task task) {

        String key = task.getTaskType()+"_"+task.getPriority();
        Calendar calendar = Calendar.getInstance();
        calendar.add(Calendar.MINUTE,5);
        long timeInMillis = calendar.getTimeInMillis();

        if (task.getExecuteTime()<=System.currentTimeMillis()){
            cacheService.lLeftPush(ScheduleConstants.TOPIC+key, JSON.toJSONString(task));
        }else if(task.getExecuteTime()<=timeInMillis){
            cacheService.zAdd(ScheduleConstants.FUTURE+key,JSON.toJSONString(task),task.getExecuteTime());
        }


    }


    @Autowired
    private TaskinfoMapper taskinfoMapper;

    @Autowired
    private TaskinfoLogsMapper taskinfoLogsMapper;
    //先保存任务表以及任务日志表  无论如何哦都得保存的    当任务被消费 修改任务状态来表示是否被消费
    private Boolean addTaskToDb(Task task) {

        boolean flag = false;

        try {
            Taskinfo taskinfo = new Taskinfo();
            BeanUtils.copyProperties(task,taskinfo);
            taskinfoMapper.insert(taskinfo);

            TaskinfoLogs taskinfoLogs = new TaskinfoLogs();
            BeanUtils.copyProperties(taskinfo,taskinfoLogs);
            taskinfoLogs.setStatus(ScheduleConstants.SCHEDULED);
            taskinfoLogs.setVersion(1);
            taskinfoLogsMapper.insert(taskinfoLogs);
            flag=true;
        }catch (Exception e){
            e.printStackTrace();
        }

        return flag;
    }


    @Scheduled(cron = "0 */1 * * * ?")//每一分钟去执行一次
    public void refresh(){

        String furute_task_sync = cacheService.tryLock("FUTURE_TASK_SYNC", 1000 * 30);

        if (StringUtils.isNotBlank(furute_task_sync)){
            log.info("未来数据定数刷新");
            Set<String> scan = cacheService.scan(ScheduleConstants.FUTURE + "*");

            for (String s : scan) {
                String topicKey = ScheduleConstants.TOPIC+s.split(ScheduleConstants.FUTURE)[1];
                Set<String> strings = cacheService.zRangeByScore(s, 0, System.currentTimeMillis());

                if (!strings.isEmpty()){
                    cacheService.refreshWithPipeline(s,topicKey,strings);
                    log.info("成功将"+s+"刷新到"+topicKey);
                }

            }
        }

    }


    @PostConstruct
    @Scheduled(cron = "0 */5 * * * ?")
    public void reloadData(){

        clearCache();
        Calendar instance = Calendar.getInstance();
        instance.add(Calendar.MINUTE,5);
        long timeInMillis = instance.getTimeInMillis();
        List<Taskinfo> taskinfos = taskinfoMapper.selectList(Wrappers.<Taskinfo>lambdaQuery().lt(Taskinfo::getExecuteTime, timeInMillis));
        if (taskinfos!=null&&taskinfos.size()>0){
            for (Taskinfo taskinfo : taskinfos) {
                Task task = new Task();
                BeanUtils.copyProperties(taskinfo,task);
                task.setExecuteTime(taskinfo.getExecuteTime().getTime());
                addTaskToCache(task);

            }
        }

        log.info("数据库的数据同步到了redis中---");

        return;
    }

    public void clearCache(){
        Set<String> scan = cacheService.scan(ScheduleConstants.TOPIC + "*");
        Set<String> scan1 = cacheService.scan(ScheduleConstants.FUTURE + "*");
        cacheService.delete(scan);
        cacheService.delete(scan1);
    }

}
